package dst.ass3.event.impl;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPAdministrator;
import com.espertech.esper.client.EPRuntime;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.StatementAwareUpdateListener;
import dst.ass3.dto.TaskWrapperDTO;
import dst.ass3.event.Constants;
import dst.ass3.event.IEventProcessing;
import dst.ass3.model.ITaskWrapper;
import dst.ass3.model.LifecycleState;

/**
 * Complex Event Processing (CEP), a technique that is becoming
 * more and more important in today's business processes and loosely coupled distributed systems. Generally
 * speaking, an "event" is anything (i.e., a phenomenon, happening or similar) that is of interest for the
 * application. CEP collectively refers to techniques for processing of event messages, including event
 * routing, event aggregation, event pattern detection, etc.
 */

public class EventProcessing implements IEventProcessing {
	
	private static final String EVENT_TASK_NAME = "TaskWrapper";
    private static final String ESPER_SERVICE_PROVIDER_NAME = "EsperEngineDST";  //You may choose an arbitrary name for the EPServiceProvider instance,e.g., "EsperEngineDST".
    private static final int WINDOW_LENGTH = 10000;
   
    private EPAdministrator esperAdministrator;
    private EPRuntime esperRuntime;
    private EPServiceProvider esperServiceProvider;

    /**
     * Initialization of Esper Configuration
     * http://esper.codehaus.org/esper-4.2.0/doc/reference/en/html/configuration.html
     */
    @Override
    public void initializeAll(StatementAwareUpdateListener listener, boolean debug) {
        Configuration config = new Configuration();
        config.addEventType(EVENT_TASK_NAME, TaskWrapperDTO.class);  //description speaks of TaskWrapper instances, but in fact you should again use the TaskWrapper DTOs
        config.addImport(LifecycleState.class);

        //Initialize the Esper platform and obtain an instance of EPServiceProvider and EPAdministrator
        this.esperServiceProvider = EPServiceProviderManager.getProvider(ESPER_SERVICE_PROVIDER_NAME, config);
        this.esperRuntime = this.esperServiceProvider.getEPRuntime();
        this.esperAdministrator = this.esperServiceProvider.getEPAdministrator();

        //Use the dynamic type definition capabilities of Esper (hint: "create schema ...") to define three new event types.
        defineTypes(listener);
        
        //Create and execute 3 EPL queries
        createQueries(listener);
    }

    
    /**
     * Use the dynamic type definition capabilities of Esper (hint: "create schema ...") to define
	 * three new event types. The names of the event types are TaskAssigned, TaskProcessed and
	 * TaskDuration
     */
    private void defineTypes(StatementAwareUpdateListener listener) {
    	String eplDefinition=null;
    	
    	//TaskAssigned
    	eplDefinition = String.format("create schema %s as (taskId long, timestamp long)",Constants.EVENT_TASK_ASSIGNED);
        this.esperAdministrator.createEPL(eplDefinition);

        eplDefinition = String.format("select * from %s", Constants.EVENT_TASK_ASSIGNED);
        this.esperAdministrator.createEPL(eplDefinition).addListener(listener);

        
        //TaskProcessed
        eplDefinition = String.format("create schema %s as (taskId long, timestamp long)",Constants.EVENT_TASK_PROCESSED);
        this.esperAdministrator.createEPL(eplDefinition);

        eplDefinition = String.format("select * from %s", Constants.EVENT_TASK_PROCESSED);
        this.esperAdministrator.createEPL(eplDefinition).addListener(listener);

        //TaskDuration
        eplDefinition = String.format("create schema %s as (taskId long, duration long)",Constants.EVENT_TASK_DURATION);
        this.esperAdministrator.createEPL(eplDefinition);

        eplDefinition = String.format("select * from %s", Constants.EVENT_TASK_DURATION);
        EPStatement statement = this.esperAdministrator.createEPL(eplDefinition);
        statement.addListener(listener);
        
        //provide a listener which outputs (to stdout) new events that result from executing the queries
        statement.addListener(new StdOutListener());

        //Event name for result of average task duration
        eplDefinition = String.format("create schema %s as (%s double)",Constants.EVENT_AVG_TASK_DURATION,Constants.EVENT_AVG_TASK_DURATION);
        this.esperAdministrator.createEPL(eplDefinition);

        eplDefinition = String.format("select * from %s", Constants.EVENT_AVG_TASK_DURATION);
        this.esperAdministrator.createEPL(eplDefinition).addListener(listener);
    }
    
	private void createQueries(StatementAwareUpdateListener listener) {
		String eplDefinition = null;

		//TaskAssigned is triggered when a TaskWrapper event with status ASSIGNED occurs
		eplDefinition = String.format("insert into %s (taskId, timestamp) "
				+ "select taskId, current_timestamp() " + "from %s "
				+ "where state = %s.%s", Constants.EVENT_TASK_ASSIGNED,
				EVENT_TASK_NAME, LifecycleState.class.getSimpleName(),
				LifecycleState.ASSIGNED.toString());
		this.esperAdministrator.createEPL(eplDefinition);

		//TaskProcessed is triggered in case of a TaskWrapper event with status PROCESSED.
		eplDefinition = String.format("insert into %s (taskId, timestamp) "
				+ "select taskId, current_timestamp() " + "from %s "
				+ "where state = %s.%s", Constants.EVENT_TASK_PROCESSED,
				EVENT_TASK_NAME, LifecycleState.class.getSimpleName(),
				LifecycleState.PROCESSED.toString());
		this.esperAdministrator.createEPL(eplDefinition);

		//TaskDuration events combine the information of the two previous task-wrapper types and contain the processing duration
		//update it with sliding window of 10000
		//You can assume that the TaskDuration query does not need to consider more than 10.000 historical events 
		// (i.e., there are never more than 10.000 events between any two correlated TaskWrapper
		eplDefinition = String.format("insert into %s (taskId, duration) "
				+ "select a.taskId, b.timestamp - a.timestamp "
				+ "from %s.win:length(%d) a, " + " %s.win:length(%d) b "
				+ "where a.taskId = b.taskId", Constants.EVENT_TASK_DURATION,
				Constants.EVENT_TASK_ASSIGNED, WINDOW_LENGTH,
				Constants.EVENT_TASK_PROCESSED, WINDOW_LENGTH);
		this.esperAdministrator.createEPL(eplDefinition);

		//Receive notifications about all events of type TaskDuration
		//emit a AvgTaskDuration event
		//computes the average execution duration over all task-wrappers that finished within the last 15 seconds
		eplDefinition = String.format("insert into %s (%s) "
				+ "select avg(duration) " + "from %s.win:time(15 seconds)",
				Constants.EVENT_AVG_TASK_DURATION,
				Constants.EVENT_AVG_TASK_DURATION,
				Constants.EVENT_TASK_DURATION);
		this.esperAdministrator.createEPL(eplDefinition);

		
		//Use pattern matching facilities of EPL to detect task-wrappers which have 3 times attempted
		//and failed to execute (i.e., switched 3 times between the state READY FOR PROCESSING and the
		//state PROCESSING NOT POSSIBLE).
		
		//http://esper.codehaus.org/esper-4.6.0/doc/reference/en-US/html/event_patterns.html#pattern-howto-syntax
		
		//If an arriving event matches the filter expression marked with @consume it is no longer available to other 
		//filter expressions of the same pattern that also match the arriving event. 
		
		//followed-by	->
		String pattern = String.format
				("every r1=%1$s(state = %2$s.%3$s) @consume(1) -> "					//Processing
						+ " %1$s(state = %2$s.%4$s, id = r1.id) @consume(2) -> "	//Not Possible
						+ " %1$s(state = %2$s.%3$s, id = r1.id) @consume(3) -> "	//Processing
						+ " %1$s(state = %2$s.%4$s, id = r1.id) @consume(4) -> "	//Not Possible
						+ " %1$s(state = %2$s.%3$s, id = r1.id) @consume(5) -> "	//Processing
						+ " %1$s(state = %2$s.%4$s, id = r1.id) @consume(6)",		//Not Possible
		EVENT_TASK_NAME, LifecycleState.class.getSimpleName(),
		LifecycleState.READY_FOR_PROCESSING.toString(),
		LifecycleState.PROCESSING_NOT_POSSIBLE.toString());

		this.esperAdministrator.createPattern(pattern).addListener(listener);	//adds a Listener when arrives
	}

    @Override
    public void addEvent(ITaskWrapper task) {
    	esperRuntime.sendEvent(new TaskWrapperDTO(task));
    }

    @Override
    public void close() {
    	esperRuntime = null;

        esperServiceProvider.destroy();
        esperServiceProvider = null;
    }

    private static class StdOutListener implements StatementAwareUpdateListener {
        @Override
        public void update(EventBean[] newEvents, EventBean[] oldEvents, EPStatement statement,EPServiceProvider epServiceProvider) {
            for (EventBean e : newEvents) {
                System.out.println(e);
            }
        }
    }

}